package com.bytes32.stats;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.io.WritableUtils;

/**
 * Implementation of {@link QBuffer} hosts
 * the aproximate quantiles and associated information
 * @author murariuf
 *
 */
final class InputBuffer implements QBuffer{
	/*
	 * Buffer holding the info 
	 */
	private double[] buf;
	/*
	 * Weight of the buffer 
	 */
	private int weight;
	/*
	 * Level of the buffer
	 */
	private int level;
	/*
	 * Status of the buffer
	 */
	private BufferStatus status = BufferStatus.EMPTY;
	/*
	 * Actual length of the buffer
	 */
	private int length = 0;
	
	/*
	 * Maximum length of the buffer after which it is considered full
	 */
	private int maxLength = 0;
	
	/*
	 * Wrapps around an existing buffer 
	 */
	InputBuffer(double[] buf, int weight, int level, BufferStatus status) {
		buf(buf);
		weight(weight);
		level(level);
		status(status);
		maxLength = buf.length;
		if (status == BufferStatus.FULL)
			length = maxLength; 
	}
	
	public InputBuffer(int maxLength, int weight, int level, BufferStatus status){
		buf(new double[maxLength]);
		weight(weight);
		level(level);
		status(status);		
		this.maxLength = maxLength;
	}

	public double get(int i) {
		return buf[i];
	}

	public BufferStatus add(final double value){
		if (length == maxLength)
			throw new IllegalStateException("Buffer is already full");
		buf[length] = value;
		length++;
		if (length == maxLength)
			return status = BufferStatus.FULL;
		return status = BufferStatus.PARTIAL;
	}
	
	public void weight(int weight) {
		this.weight = weight;
	}

	public int weight() {
		return this.weight;
	}

	public void level(int level) {
		this.level = level;
	}

	public int level() {
		return this.level;
	}

	public void status(BufferStatus status) {
		this.status = status;
	}

	public BufferStatus status() {
		return this.status;
	}

	void buf(double[] buf) {
		this.buf = buf.clone();
	}

	@Override
	public String toString() {
		return String.format("w : %d, l: %d, status: %s, buf : %s", weight,
				level, status.name(), Arrays.toString(buf));
	}

	public int getLength() {
		return length;
	}

	public int getMaxLength() {
		return maxLength;
	}
	
	/**
	 * Sorts the array, ignores the empty zeroes
	 */
	public void sort(){
		double [] temp = Arrays.copyOf(this.buf, this.length);
		Arrays.sort(temp);
		this.buf = Arrays.copyOfRange(temp, 0, this.maxLength);
	}
	
	public void clear(){
		clearArray();
		this.length = 0;
		this.weight = 0;
		this.level = 0;
	}

	private void clearArray(){
		Arrays.fill(this.buf,0);
	}
	
	public double[] toArray(){
		return this.buf.clone();
	}

	@Override
	public void readFields(DataInput stream) throws IOException {
		this.length = WritableUtils.readVInt(stream);
		this.maxLength = WritableUtils.readVInt(stream);
		this.level = WritableUtils.readVInt(stream);
		this.weight = WritableUtils.readVInt(stream);
		this.buf = new double[this.maxLength];
		for (int i = 0; i < this.length; i++){
			this.buf[i] = Double.longBitsToDouble(WritableUtils.readVLong(stream));
		}
	}

	@Override
	public void write(DataOutput stream) throws IOException {
		WritableUtils.writeVInt(stream, this.length);
		WritableUtils.writeVInt(stream, this.maxLength);
		WritableUtils.writeVInt(stream, this.level);
		WritableUtils.writeVInt(stream, this.weight);
		for (int i = 0; i< this.length; i++){
			WritableUtils.writeVLong(stream, Double.doubleToLongBits(this.buf[i]));
		}
	}
}
